package cn.ac.iie.di.cst.lls.worker;

/*
 * To change this template, choose Tools | Templates
 * and open the template in the editor.
 */
import cn.ac.iie.di.cst.lls.common.RuntimeEnv;
import static cn.ac.iie.di.cst.lls.common.RuntimeEnv.docsSchema;
import static cn.ac.iie.di.cst.lls.common.RuntimeEnv.mqschema;
import cn.ac.iie.di.cst.lls.util.RocketMQProducer;
import java.io.ByteArrayOutputStream;
import static java.lang.Thread.sleep;
import java.nio.ByteBuffer;
import org.apache.avro.Schema;
import org.apache.avro.generic.GenericArray;
import org.apache.avro.generic.GenericData;
import org.apache.avro.generic.GenericDatumWriter;
import org.apache.avro.generic.GenericRecord;
import org.apache.avro.io.BinaryEncoder;
import org.apache.avro.io.DatumWriter;
import org.apache.avro.io.EncoderFactory;
import org.apache.log4j.Logger;

/**
 *
 * @author
 */
public class SendMsgToMq implements Runnable {
    
    Schema docSchema = null;
    String docschemaname = null;
    String productmq = null;
//    int i = 0;
    long lastUpdateTime = -1;
    long timeout = -1;
    
    private static org.apache.log4j.Logger logger = Logger.getLogger(SendMsgToMq.class.getName());
    
    public SendMsgToMq(String mq) {
        docSchema = mqschema.get(mq);
        docschemaname = mq;
        productmq = mq + "_mq";
    }
    
    @Override
    public void run() {
        if (RuntimeEnv.mq2queue.get(docschemaname).equals("")) {
            logger.info(" SEND  RuntimeEnv.mq2queue.get(docschemaname) == null and exit");
            System.exit(0);
        }
        logger.info("init send mq successfully" + docschemaname);
        while (true) {
            try {
                long tt = System.currentTimeMillis();
                DatumWriter<GenericRecord> docsWriter = new GenericDatumWriter<GenericRecord>(docsSchema);
                ByteArrayOutputStream docssos = new ByteArrayOutputStream();
                BinaryEncoder docsbe = new EncoderFactory().binaryEncoder(docssos, null);
                GenericRecord docsRecord = new GenericData.Record(docsSchema);
                int bs = Integer.valueOf(RuntimeEnv.getParam(RuntimeEnv.BatchSize).toString());
                GenericArray docSet = new GenericData.Array<ByteBuffer>(bs, docsSchema.getField("doc_set").schema());
                DatumWriter<GenericRecord> dWriter = new GenericDatumWriter<GenericRecord>(docSchema);
                ByteArrayOutputStream bos = new ByteArrayOutputStream();
                BinaryEncoder be = new EncoderFactory().binaryEncoder(bos, null);
                
                for (int i = 0; i < bs; i++) {
                    GenericRecord docRecord = null;
                    while (true) {
                        docRecord = (GenericRecord) RuntimeEnv.mq2queue.get(docschemaname).poll();
                        if (docRecord != null) {
                            break;
                        } else {
                            sleep(500);
                            if (docSet.size() != 0 && RuntimeEnv.mq2queue.get(docschemaname).size() == 0) {
                                logger.info(docschemaname + "is end");
                                break;
                            }
                        }
                    }
                    if (docRecord == null) {
//                        logger.error("docRecord is null");
                        break;
                    }
                    dWriter.write(docRecord, be);
                    be.flush();
                    docSet.add(ByteBuffer.wrap(bos.toByteArray()));
                    bos.reset();
                    i += bos.toByteArray().length;
                }
//                logger.info("will send "+docSet.size()+" message ");
                docsRecord.put("doc_set", docSet);
                docsRecord.put("sign", "your sign kong");
                docsRecord.put("doc_schema_name", docschemaname);
                
                docsWriter.write(docsRecord, docsbe);
                docsbe.flush();
                
                long endt = System.currentTimeMillis() / 1000 - tt / 1000;
                RocketMQProducer.sendMessage(productmq, docssos.toByteArray());
                logger.info("nowtime:" + System.currentTimeMillis() / 1000 + " SEND  " + docschemaname + " batchsize " + docSet.size() + "  use  " + endt + "s time");
                docSet.clear();

//                docSet = new GenericData.Array<ByteBuffer>(2000, docsSchema.getField("doc_set").schema());
//                i = 0;
            } catch (Exception ex) {
                logger.error(ex);
            }
        }
    }
}